/**
 * Created by IntelliJ IDEA.
 * User: jiyufei
 * Date: 2018/10/17
 * Time: 13:55
 **/
package com.jiyufei.linux.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class SwitchBolt extends BaseRichBolt {
    
    private OutputCollector outputCollector;
    private TopologyContext topologyContext;
    private Pattern pattern;
    private Matcher m;
    private String type;
    
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputCollector = outputCollector;
        this.topologyContext = topologyContext;
        pattern = Pattern.compile("(\\S+)\\s(.+)");
    }
    
    @Override
    public void execute(Tuple tuple) {
        String log = tuple.getString(0);
        m = pattern.matcher(log);
        if(m.find()){
            type = m.group(1);
            switch (type){
                case "secure":
                    outputCollector.emit("secure-stream",tuple,new Values(log));
                    outputCollector.ack(tuple);
                    break;
                case "message":
                    outputCollector.emit("message-stream",tuple,new Values(log));
                    outputCollector.ack(tuple);
                    break;
                case "cron":
                    outputCollector.emit("cron-stream",tuple,new Values(log));
                    outputCollector.ack(tuple);
                    break;
                case "yum":
                    outputCollector.emit("yum-stream",tuple,new Values(log));
                    outputCollector.ack(tuple);
                    break;
                default:
                    outputCollector.emit("other-stream",tuple,new Values(log));
                    outputCollector.ack(tuple);
                    break;
            }
        }
    }
    
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declareStream("secure-stream",new Fields("log"));
        outputFieldsDeclarer.declareStream("message-stream",new Fields("log"));
        outputFieldsDeclarer.declareStream("cron-stream",new Fields("log"));
        outputFieldsDeclarer.declareStream("yum-stream",new Fields("log"));
        outputFieldsDeclarer.declareStream("other-stream",new Fields("log"));
    }
}
